package com.fzk.ha;

import com.fzk.core.FileStoreService;
import com.fzk.env.conf.DataConf;
import com.fzk.env.conf.ServerConf;
import com.fzk.log.Logger;
import com.fzk.server.NettyHttpServer;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 主从复制
 */
public class HAService implements AutoCloseable {
    public static final String Master = "master";
    public static final String Replica = "replica";
    private final NettyHttpServer server;
    private final FileStoreService fileService;
    private final ServerConf serverConf;
    private final DataConf dataConf;
    private final InetSocketAddress self;
    // 节点集合，包含自身
    private final HashSet<InetSocketAddress> replicas = new HashSet<>();
    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    // 主节点：保存连上的从节点的channel
    // 从节点：保存与主节点的连接channel
    public final ConcurrentHashMap<ChannelId, Channel> channels = new ConcurrentHashMap<>();

    public HAService(NettyHttpServer server, FileStoreService fileService, ServerConf serverConf, DataConf dataConf) {
        this.server = server;
        this.fileService = fileService;
        this.serverConf = serverConf;
        this.dataConf = dataConf;
        String[] splits = this.serverConf.getReplicaStr().split(",s*");
        for (String split : splits) {
            String[] sp = split.split(":");
            if (sp.length != 2) {
                throw new RuntimeException("配置错误，非法地址：" + serverConf.getReplicaStr());
            }
            InetSocketAddress address = new InetSocketAddress(sp[0], Integer.parseInt(sp[1]));
            this.replicas.add(address);
        }
        this.self = new InetSocketAddress("127.0.0.1", this.serverConf.getHaPort());
        this.replicas.add(this.self);// HashSet保证了不会重复

        this.bossGroup = new NioEventLoopGroup(1);
        this.workerGroup = new NioEventLoopGroup(replicas.size());
    }

    public void start() {
        final HAService haService = this;
        // 1、每个replica都监听18080端口，获取来自其它replica的连接

        // Configure SSL.
        // final SslContext sslCtx = ServerUtil.buildSslContext();
        // Configure the server.
        if (Master.equals(this.serverConf.getRole())) {
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline p = ch.pipeline();
                                // if (sslCtx != null) {
                                // p.addLast(sslCtx.newHandler(ch.alloc()));
                                // }
                                // p.addLast(new LoggingHandler(LogLevel.INFO));
                                // 链路空闲超时控制：服务端监听读空闲
                                p.addLast(new IdleStateHandler(serverConf.getHaIdleTimeOut(), 0, 0));
                                /*
                                 * 消息帧最大100MB
                                 * 长度字段偏移量为0，4B
                                 * 无需长度修正，无需舍弃开头的长度字段
                                 */
                                LengthFieldBasedFrameDecoder frameDecoder = new LengthFieldBasedFrameDecoder(
                                        1024 * 1024 * 100, 0, 4, 0, 0);
                                p.addLast(frameDecoder);// TCP拆包解包
                                p.addLast(new ServerHandler(haService, fileService, serverConf, dataConf));

                            }
                        });

                // Start the server.
                ChannelFuture f = b.bind(this.serverConf.getHaPort()).sync();

                // Wait until the server socket is closed.
                //f.channel().closeFuture().sync();
            } catch (Exception e) {
                // Shut down all event loops to terminate all threads.
                close();
                e.printStackTrace();
                Logger.error(e.toString());
            }
        }


        // 2、每个replica都主动去连接master
        if (Replica.equals(serverConf.getRole())) {
            String[] splits = serverConf.getMasterStr().split(":");
            if (splits.length != 2) {
                throw new RuntimeException("非法主节点配置: " + serverConf);
            }
            InetSocketAddress replica = new InetSocketAddress(splits[0], Integer.parseInt(splits[1]));
//            for (InetSocketAddress replica : replicas) {
            if (!self.equals(replica)) {
                try {
                    // 创建Bootstrap客户端启动器
                    Bootstrap b = new Bootstrap();
                    b.group(workerGroup)
                            .channel(NioSocketChannel.class)    // 设置Channel类型
                            .option(ChannelOption.TCP_NODELAY, true)    // 关闭TCP延迟
                            .handler(new ChannelInitializer<SocketChannel>() {
                                @Override
                                public void initChannel(SocketChannel ch) throws Exception {
                                    ChannelPipeline p = ch.pipeline();
                                    //if (sslCtx != null) {
                                    //    p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
                                    //}
                                    //p.addLast(new LoggingHandler(LogLevel.INFO));
                                    // 链路空闲超时控制：客户端监听读空闲
                                    p.addLast(new IdleStateHandler(serverConf.getHaIdleTimeOut(), 0, 0));
                                    /*
                                     * 消息帧最大100MB
                                     * 长度字段偏移量为0，4B
                                     * 无需长度修正，无需舍弃开头的长度字段
                                     */
                                    LengthFieldBasedFrameDecoder frameDecoder = new LengthFieldBasedFrameDecoder(
                                            1024 * 1024 * 100, 0, 4, 0, 0);
                                    p.addLast(frameDecoder);// TCP拆包解包
                                    p.addLast(new ServerHandler(haService, fileService, serverConf, dataConf));// 添加自定义处理器
                                }
                            });

                    // 连接服务器并等待
                    ChannelFuture f = b.connect(replica).sync();
                    // 定时发送心跳
//                    f.channel().eventLoop().scheduleWithFixedDelay(() -> {
//                        HAMsg.HAHeartMsg heartMsg = new HAMsg.HAHeartMsg(HAMsg.Ping, ("hello i am client: " + self).getBytes(StandardCharsets.UTF_8));
//                        f.channel().writeAndFlush(heartMsg.serialize());
//                    }, 3, serverConf.getHeartInternal(), TimeUnit.SECONDS);

                    // 发HA sync同步请求
                    f.channel().eventLoop().schedule(() -> {
                        Logger.debug("第一次HA sync req");
                        HAMsg.HASyncRequestMsg req = new HAMsg.HASyncRequestMsg(new HAMsg.HASyncRequestHeader(HAMsg.HASyncRequestHeader.NormalSync, fileService.metaStore.writeIndex, 0));
                        f.channel().writeAndFlush(req.serialize());
                    }, 2, TimeUnit.SECONDS);
                    // Wait until the connection is closed.
                    //f.channel().close().sync(); // 手动关闭通道并等待
                    //f.channel().closeFuture().sync(); // 等待直至通道关闭
                } catch (Exception e) {
                    // Shut down the event loop to terminate all threads.
                    close();
                    e.printStackTrace();
                    Logger.error(e.toString());
                }
//            }
            }
        }
    }

    @Override
    public void close() {
        this.bossGroup.shutdownGracefully();
        this.workerGroup.shutdownGracefully();
        this.channels.clear();
    }
}
